package edu.neu.swc.skynet.service.async;


import com.alibaba.fastjson.JSON;
import edu.neu.swc.skynet.dao.MovieDocRepository;
import edu.neu.swc.skynet.dao.SkyNetMovieDao;
import edu.neu.swc.skynet.document.MovieDoc;
import edu.neu.swc.skynet.entity.SkyNetMovie;
import edu.neu.swc.skynet.filter.AccessContext;
import edu.neu.swc.skynet.movie.MovieInfo;
import edu.neu.swc.skynet.vo.LoginUserInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.IterableUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@Transactional
public class AsyncServiceImpl implements IAsyncService{

    private final SkyNetMovieDao skyNetMovieDao;

    private final MovieDocRepository movieDocRepository;

    public AsyncServiceImpl(SkyNetMovieDao skyNetMovieDao,
                            MovieDocRepository movieDocRepository) {
        this.skyNetMovieDao = skyNetMovieDao;
        this.movieDocRepository = movieDocRepository;
    }

    /**
     * <h2>异步任务需要加上注解，指定使用的线程池</h2>
     * 异步任务处理两件事
     * 1，将商品信息保存到数据表
     * 2，更新商品缓存
     */
    @Override
    @Async("getAsyncExecutor")
    public void asyncImportMovie(List<MovieInfo> movieInfos, String taskId) {
        log.info("async task running taskId[{}]",taskId);

        StopWatch watch = StopWatch.createStarted();

        //列表批量转换 MovieInfo -> SkyNetMovie
        List<SkyNetMovie> targetMovies = movieInfos.stream()
                .map(SkyNetMovie::to)
                .collect(Collectors.toList());



        //异步导入数据库
        List<SkyNetMovie> savedMovies = IterableUtils.toList(
                skyNetMovieDao.saveAll(targetMovies)
        );

        List<MovieDoc> movieDocs = savedMovies.stream()
                .map(SkyNetMovie::toMovieDoc)
                .collect(Collectors.toList());

        //异步导入 ES
        log.info("import to ES:[{}]",JSON.toJSONString(
                        IterableUtils.toList(
                                movieDocRepository.saveAll(movieDocs)
                        )
                )
        );

        log.info("save movies to DB:[{}]",
                JSON.toJSONString(savedMovies));
        watch.stop();

        log.info("check and import goods success:[{}ms]", watch.getTime(TimeUnit.MILLISECONDS));
    }

    /**
     * <h2>数据同步实现</h2>
     * 1. 从 DB 中获取数据
     * 2. 数据类型转换
     * 3. 清空 ES
     * 4. 导入 ES 中
     * @param taskId 分派的异步任务 ID
     * @param placeHolder 占位符
     */
    @Override
    @Async("getAsyncExecutor")
    public void syncMovieShow(String placeHolder, String taskId) {

        //TODO: 需要检查权限信息，该用户是否拥有权限异步导入
        LoginUserInfo loginUserInfo = AccessContext.getLoginUserInfo();
        //TODO: 远程访问用户信息，判断是否拥有权限
        //TODO: 如果拥有，放行
        //TODO: 如果没有，抛出异常
        log.info("async task running taskId[{}]",taskId);
        StopWatch watch = StopWatch.createStarted();
        log.info("placeHolder:[{}]", placeHolder);

        // 1.从 DB 中获取数据
        List<SkyNetMovie> skyNetMovies = IterableUtils.toList(
                skyNetMovieDao.findAll()
        );

        // 2.转换成 MovieDoc 类型
        List<MovieDoc> movieDocs = skyNetMovies.stream()
                .map(SkyNetMovie::toMovieDoc)
                .collect(Collectors.toList());

        // 3.清空 ES, 为了保证强一致性
        movieDocRepository.deleteAll();

        // 4.导入 ES 数据
        log.info("import DB data to ES:[{}]",
                JSON.toJSONString(
                        movieDocRepository.saveAll(movieDocs)
                ));

        watch.stop();
        log.info("check and import goods success:[{}ms]", watch.getTime(TimeUnit.MILLISECONDS));
    }
}
